﻿//////////////////////////////////////////////////////////////////
//
// Rabbit\Transport.cs
//
// Copyright (c) 2012 Dan Pike.
//
// Permission is hereby granted, free of charge, to any person
// obtaining a copy of this software and associated documentation files
// (the "Software"), to deal in the Software without restriction,
// including without limitation the rights to use, copy, modify, merge,
// publish, distribute, sublicense, and/or sell copies of the Software,
// and to permit persons to whom the Software is furnished to do so,
// subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included
// in all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
// IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
// CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
// TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
// SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
//
using Interfaces;
using ProtoBuf;
using RabbitMQ;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Util;
using System;
using System.Collections.Generic;
using System.Configuration;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading;
using Utilities;
using RabbitMQ.Client.Framing.v0_9_1;

namespace Rabbit
{
   public class Transport : ITransport
   {
      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// The DateTime.Ticks value corresponding to time_t == 0 in Unix
      /// </summary>
      private static long EPOQUE;

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// Implementation of ITransport events
      /// </summary>
      public event NetworkStateChangedHandler EventStateChanged;
      public event ReceivedMessageHandler EventReceivedMessage;

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// Private properties
      /// </summary>
      private static Logger logger_;
      private int streamSize_;
      private string senderId_;
      private IAccount account_;
      private Thread threadRabbit_;
      private DateTime threadRetryConnectAt_;
      private EventWaitHandle signalStop_;
      private AmqpUriConnectionFactory factory_;
      private IConnection connection_;
      private IModel channel_;
      private string queueName_;
      private QueueingBasicConsumer consumer_;

      // Use the state property to get/set this value as it will broadcast the
      // correct events to update the GUI etc
      private NetworkStateConstants state_DONTUSEDIRECTLY_ = NetworkStateConstants.None;

      /// <summary>
      /// The names of all of the exchanges that we want to bind to
      /// </summary>
      private Dictionary<string, Dictionary<string, bool>> wantExchangeNames_;
      private Dictionary<string, HashSet<string>> exchangeNames_;
      private EventWaitHandle wantPending_;

      /// <summary>
      /// Messages that need to be sent to RabbitMQ
      /// </summary>
      private class PostQueueContent
      {
         public IMessage msg;
         public Type type;
         public string exchange;
         public string routingKey;
      }
      private Queue<PostQueueContent> postQueue_;

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// 
      /// </summary>
      static Transport()
      {
         EPOQUE = new DateTime(1970, 1, 1).Ticks;
      }

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// 
      /// </summary>
      public Transport()
      {
         signalStop_ = new EventWaitHandle(false, EventResetMode.ManualReset);
         streamSize_ = 64;
         wantExchangeNames_ = new Dictionary<string, Dictionary<string, bool>>();
         wantPending_ = new EventWaitHandle(true, EventResetMode.ManualReset);
         exchangeNames_ = new Dictionary<string, HashSet<string>>();
         postQueue_ = new Queue<PostQueueContent>();
      }

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// Implementation of ITransport
      /// </summary>
      public bool Initialize(string senderId)
      {
         try
         {
            // DON'T DO THIS IN THE CONSTRUCTOR, IT HASN'T BEEN INITIALIZED UNTIL NOW
            logger_ = Program.Logger;

            logger_.DebugFormat("initializing Rabbit.Network");
            senderId_ = senderId;
            
            streamSize_ = Program.Settings.StreamSize;    // Starting size for the MemoryStream that we use to encode messages

            logger_.DebugFormat("starting Rabbit.Network thread");
            threadRabbit_ = new Thread(ThreadRabbit);
            threadRabbit_.Name = "Rabbit.Network";
            signalStop_.Reset();
            threadRabbit_.Start();
            return true;
         }
         catch (System.Exception ex)
         {
            logger_.Exception(ex);
         }
         return false;
      }

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// Implementation of ITransport
      /// </summary>
      public void Shutdown()
      {
         try
         {
            signalStop_.Set();
            if (null != threadRabbit_)
            {
               while (threadRabbit_.IsAlive)
               {
                  if (!threadRabbit_.Join(3000))
                  {
                     logger_.WarnFormat("still waiting for rabbit thread to shut down");
                  }
               }
            }
            logger_.WarnFormat("Rabbit.Network has shut down");
         }
         catch (System.Exception ex)
         {
            logger_.Exception(ex);
         }
      }

      ////////////////////////////////////////////////////////////////////////

      private NetworkStateConstants State
      {
         get { return state_DONTUSEDIRECTLY_; }
         set
         {
            switch (value)
            {
               case NetworkStateConstants.Idle:
                  // Shut down the rabbit connection
                  consumer_ = null;
                  if (null != channel_)
                  {
                     if (null != queueName_)
                     {
                        channel_.QueueUnbind(queueName_, ExchangeNames.Broadcast, string.Empty, null);
                        queueName_ = null;
                     }
                     channel_.Close();
                     channel_.Dispose();
                     channel_ = null;
                  }
                  if (null != connection_)
                  {
                     connection_.Close();
                     connection_.Dispose();
                     connection_= null;
                  }
                  break;

               default:
                  break;
            }

            if (state_DONTUSEDIRECTLY_ != value)
            {
               state_DONTUSEDIRECTLY_ = value;
               if (null != EventStateChanged)
               {
                  NetworkStateChangedEventArgs eventArgs = new NetworkStateChangedEventArgs();
                  eventArgs.state = value;
                  EventStateChanged(this, eventArgs);
               }
            }
         }
      }

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// Implementation of ITransport
      /// </summary>
      public bool Connect(IAccount account)
      {
         try
         {
            account_ = account;
            return true;
         }
         catch (System.Exception ex)
         {
            logger_.Exception(ex);
         }
         return false;
      }

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// Implementation of ITransport
      /// </summary>
      public bool Disconnect(IAccount account)
      {
         try
         {
            account_ = null;
            return true;
         }
         catch (System.Exception ex)
         {
            logger_.Exception(ex);
         }
         return false;
      }

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// Returns the AMQP URL built from the configuration settings
      /// </summary>
      /// <returns></returns>
      private string BuildServerUrl()
      {
         try
         {
            StringBuilder sb = new StringBuilder("amqp://");
            if (!string.IsNullOrWhiteSpace(Program.Settings.UserName))
            {
               sb.Append(Program.Settings.UserName);

               string password = Program.Settings.Password;
               if (!string.IsNullOrWhiteSpace(password))
               {
                  sb.Append(':');
                  sb.Append(password);
               }
               sb.Append('@');
            }
            sb.Append(Program.Settings.Address);
            
            // Port is optional
            int port = Program.Settings.Port;
            if (0 != port)
            {
               sb.Append(':');
               sb.Append(port);
            }

            // Vhost is also optional
            string vhost = Program.Settings.Vhost;
            if (!vhost.StartsWith("/"))
            {
               sb.Append('/');
            }
            sb.Append(vhost);

            return sb.ToString();
         }
         catch (System.Exception ex)
         {
            logger_.Exception(ex);
         }
         return string.Empty;
      }

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// Implementation of ITransport
      /// </summary>
      public bool LoadTile(string planetName, ILocation location)
      {
         ILoadTile message = Program.Loader.Create<ILoadTile>();
         if (null == message)
         {
            return false;
         }
         message.PlanetName = planetName;
         message.Location = location.Clone();

         string exchangeName = string.Concat(ExchangeNames.Planet, planetName);
         return Post<ILoadTile>(message, exchangeName, "*");
      }

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// Implementation of ITransport
      /// </summary>
      public bool TileImage(string planetName, ILocation location, short[,] heights)
      {
         ITileImage message = Program.Loader.Create<ITileImage>();
         if (null == message)
         {
            return false;
         }
         message.PlanetName = planetName;
         message.Location = location.Clone();
         message.Heights = heights;

         string exchangeName = string.Concat(ExchangeNames.Planet, planetName);
         return Post<ITileImage>(message, exchangeName, "*");
      }

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// Implementation of ITransport
      /// </summary>
      public bool Broadcast<T>(IMessage msg)
         where T : IMessage
      {
         return Post<T>(msg, ExchangeNames.Broadcast, string.Empty);
      }

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// Send the message on a specific routing key and exchange
      /// </summary>
      private bool Post<T>(IMessage msg, string exchange, string routingKey)
         where T : IMessage
      {
         try
         {
            PostQueueContent content = new PostQueueContent();
            content.msg = msg;
            content.type = typeof(T);
            content.exchange = exchange;
            content.routingKey = routingKey;
            lock (postQueue_)
            {
               postQueue_.Enqueue(content);
            }
         }
         catch (System.Exception ex)
         {
            logger_.Exception(ex);
         }
         return true;
      }

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// Send the message on a specific routing key and exchange
      /// </summary>
      private bool ThreadSend(DateTime now)
      {
         try
         {
            if ((null != account_) && (null != channel_))
            {
               // Grab a message, if there is one waiting
               PostQueueContent content = null;
               lock (postQueue_)
               {
                  if (0 < postQueue_.Count)
                  {
                     content = postQueue_.Dequeue();
                  }
               }

               if (null != content)
               {
                  // Package up the message and send it
                  BasicProperties properties = new BasicProperties();
                  properties.AppId = "Charles";
                  properties.UserId = account_.UserName;
                  properties.ContentEncoding = "binary/Interfaces.IMessage";
                  properties.ContentType = content.type.FullName;
                  properties.ReplyTo = senderId_;
                  properties.Timestamp = new RabbitMQ.Client.AmqpTimestamp((now.Ticks - EPOQUE) / 10000000L);

                  MemoryStream stream = new MemoryStream(streamSize_);
                  content.msg.SenderId = senderId_;
                  content.msg.Timestamp = now;
                  content.msg.Serialize(stream);
                  byte[] rawBytes = stream.GetBuffer();
                  channel_.BasicPublish(content.exchange, content.routingKey, true, true, properties, rawBytes);
                  return true;
               }
            }
         }
         catch (System.Exception ex)
         {
            logger_.Exception(ex);
            ThreadDisconnect();
         }
         return false;
      }

      //////////////////////////////////////////////////////////////////////////
      /// <summary>
      ///
      /// </summary>
      private void SubscribeExchange(string exchangeName, string routingKey,
         bool subscribe)
      {
         // Register to receive notifications on the given exchange
         lock (wantExchangeNames_)
         {
            // Throw it over to the Rabbit thread for processing
            Dictionary<string, bool> routingKeys;
            if (!wantExchangeNames_.TryGetValue(exchangeName, out routingKeys))
            {
               routingKeys = new Dictionary<string, bool>();
               wantExchangeNames_.Add(exchangeName, routingKeys);
            }
            if (routingKeys.ContainsKey(routingKey))
            {
               routingKeys[routingKey] = subscribe;
            }
            else
            {
               routingKeys.Add(routingKey, subscribe);
            }
            wantPending_.Reset();
         }
      }

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// Make sure that we are connected and that all pending setup operations
      /// (exchange connections etc) have been completed
      /// </summary>
      /// <returns>true if connected and stable</returns>
      private bool EnsureConnected()
      {
         try
         {
            // Wait for all of the pending exchange actions to be completed
            if (wantPending_.WaitOne(5000))
            {
               return true;
            }
            logger_.ErrorFormat("timed out waiting for exchange operations to complete");
         }
         catch (System.Exception ex)
         {
            logger_.Exception(ex);
         }
         return false;
      }

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// Implementation of ITransport
      /// </summary>
      public bool OpenPlanet(string planet)
      {
         try
         {
            logger_.DebugFormat("opening connection to planet '{0}'", planet);

            // Cancel any previous connection to this planet            
            if (!ClosePlanet(planet))
            {
               return false;
            }

            // Start a new connection
            string exchangeName = string.Concat(ExchangeNames.Planet, planet);
            SubscribeExchange(exchangeName, "*", true);

            // Wait for it to complete
            return EnsureConnected();
         }
         catch (System.Exception ex)
         {
            logger_.Exception(ex);
         }
         return false;
      }

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// Implementation of ITransport
      /// </summary>
      public bool ClosePlanet(string planet)
      {
         try
         {
            string exchangeName = string.Concat(ExchangeNames.Planet, planet);
            SubscribeExchange(exchangeName, "*", false);

            // Wait for it to complete
            return EnsureConnected();
         }
         catch (System.Exception ex)
         {
            logger_.Exception(ex);
         }
         return false;
      }

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// Implementation of ITransport
      /// </summary>
      public bool StartPlanet(string planet)
      {
         try
         {
            logger_.DebugFormat("starting planet '{0}'", planet);

            // Cancel any previous associations with this planet            
            if (!StopPlanet(planet))
            {
               return false;
            }

            // Subscribe to requests for this planet
            string exchangeName = string.Concat(ExchangeNames.Planet, planet);
            SubscribeExchange(exchangeName, "*", true);

            // Wait for it to complete
            if (!EnsureConnected())
            {
               return false;
            }
            return true;
         }
         catch (System.Exception ex)
         {
            logger_.Exception(ex);
         }
         return false;
      }

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// Implementation of ITransport
      /// </summary>
      public bool StopPlanet(string planet)
      {
         try
         {
            // and stop listening for actions
            string exchangeName = string.Concat(ExchangeNames.Planet, planet);
            SubscribeExchange(exchangeName, "*", false);

            // Wait for it to complete
            return EnsureConnected();
         }
         catch (System.Exception ex)
         {
            logger_.Exception(ex);
         }
         return false;
      }

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// 
      /// </summary>
      private void ThreadRabbit()
      {
         bool done = false;
         try
         {
            logger_.DebugFormat("rabbit thread is starting");
            threadRetryConnectAt_ = DateTime.Now;
            State = NetworkStateConstants.Idle;

            bool didWork = false;
            while (!done)
            {
               int sleepy = didWork ? 0 : 100;
               if (signalStop_.WaitOne(sleepy))
               {
                  logger_.DebugFormat("rabbit thread detected request to stop");
                  done = true;
                  break;
               }

               didWork = false;
               DateTime now = DateTime.Now;
               switch (State)
               {
                  case NetworkStateConstants.Idle:
                     // Do we want to retry connecting to Rabbit?
                     didWork = ThreadConnect();
                     break;

                  case NetworkStateConstants.Connected:
                     // Should we connect to any exchanges?
                     didWork = ThreadPlanet();
                     
                     // should we send any messages?
                     if (!didWork)
                     {
                        didWork = ThreadSend(now);
                     }

                     // Have we received a message?
                     if (!didWork)
                     {
                        didWork = ThreadResponse();
                     }
                     break;
               }
            }
            logger_.DebugFormat("rabbit thread is stopping");

            // Shut down the rabbit connection
            State = NetworkStateConstants.Idle;
         }
         catch (System.Exception ex)
         {
            logger_.Exception(ex);
         }
         finally
         {
            logger_.DebugFormat("rabbit thread is exiting");
         }
      }

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// 
      /// </summary>
      private bool ThreadConnect()
      {
         string serverUrl = string.Empty;
         try
         {
            if (DateTime.Now < threadRetryConnectAt_)
            {
               // Too early since the previous failure
               return false;
            }

            serverUrl = BuildServerUrl();
            factory_ = new AmqpUriConnectionFactory(serverUrl);
            if (!factory_.validUri)
            {
               logger_.ErrorFormat("Badly-formed AMQP string, it must follow the format: "
                  + "'amqp://guest:guest@localhost:5672/', note the trailing '/' is required!");
               return false;
            }
            connection_ = factory_.CreateConnection();
            channel_ = connection_.CreateModel();
            channel_.BasicReturn += new BasicReturnEventHandler(ThreadHandleBasicReturn);

            queueName_ = channel_.QueueDeclare();
            consumer_ = new QueueingBasicConsumer(channel_);
            channel_.BasicConsume(queueName_, true, consumer_);
            channel_.ExchangeDeclare(ExchangeNames.Broadcast, ExchangeType.Fanout, false, true, null);
            channel_.QueueBind(queueName_, ExchangeNames.Broadcast, string.Empty);

            logger_.Debug("now connected to rabbit");
            State = NetworkStateConstants.Connected;
            return true;
         }
         catch (System.Exception ex)
         {
            logger_.Exception("failed to connect to Rabbit with url '{0}'", ex, serverUrl);
            ThreadDisconnect();
         }
         return false;
      }

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// 
      /// </summary>
      private void ThreadDisconnect()
      {
         // Shut down everything and try to reconnect
         try
         {
            logger_.DebugFormat("dropping rabbit connection");
            threadRetryConnectAt_ = DateTime.Now + new TimeSpan(0, 0, 10);
            State = NetworkStateConstants.Idle;

            if (null != channel_)
            {
               channel_.QueueUnbind(queueName_, ExchangeNames.Broadcast, string.Empty, null);

               // Drop the existing bindings
               List<KeyValuePair<string, HashSet<string>>> exchanges = exchangeNames_.ToList();
               exchangeNames_.Clear();
               exchanges.ForEach(kvp =>
               {
                  string exchangeName = kvp.Key;
                  HashSet<string> routingKeys = kvp.Value;

                  // Drop our connection to these exchange/routing key combinations
                  // and put them back on the wanted list
                  routingKeys.ToList().ForEach(routingKey =>
                     {
                        try
                        {
                           channel_.QueueUnbind(queueName_, exchangeName, routingKey, null);
                           SubscribeExchange(exchangeName, routingKey, true);
                        }
                        catch (System.Exception ex)
                        {
                           logger_.Exception(ex);
                        }
                     });
               });
            }
            if (null != consumer_)
            {
               consumer_.Queue.Close();
               consumer_ = null;
            }
            if (null != channel_)
            {
               channel_.Close();
               channel_.Dispose();
               channel_ = null;
            }
            if (null != connection_)
            {
               connection_.Close();
               connection_.Dispose();
               connection_ = null;
            }
            factory_ = null;
         }
         catch (System.Exception ex)
         {
            logger_.Exception(ex);
         }
      }

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// Bind to the given exchange/routingKey combination
      /// </summary>
      private bool ThreadBindRoutingKey(string exchangeName, string routingKey)
      {
         HashSet<string> activeKeys;
         try
         {
            // Get the list of routing keys that are already bound to this exchange (if any)
            if (!exchangeNames_.TryGetValue(exchangeName, out activeKeys))
            {
               activeKeys = new HashSet<string>();
               exchangeNames_.Add(exchangeName, activeKeys);
            }
            // Add this key and, if it isn't already in there, we bind it to our queue
            if (activeKeys.Add(routingKey))
            {
               logger_.DebugFormat("binding to exchange '{0}', routing-key '{1}'",
                  exchangeName, routingKey);
               channel_.QueueBind(queueName_, exchangeName, routingKey, null);
            }
            else
            {
               logger_.DebugFormat("ignoring extraneous request to bind to exchange '{0}', routing-key '{1}'",
                  exchangeName, routingKey);
            }
            return true;
         }
         catch (System.Exception ex)
         {
            logger_.Exception(ex);
         }
         return false;
      }

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// Unbind from the given exchange/routingKey combination
      /// </summary>
      private bool ThreadUnbindRoutingKey(string exchangeName, string routingKey)
      {
         HashSet<string> activeKeys;
         try
         {
            if (exchangeNames_.TryGetValue(exchangeName, out activeKeys) &&
               activeKeys.Contains(routingKey))
            {
               // This routing key is bound to the queue, so we drop it
               logger_.DebugFormat("dropping binding to exchange '{0}', routing-key '{1}'",
                  exchangeName, routingKey);
               channel_.QueueUnbind(queueName_, exchangeName, routingKey, null);
            }
            else
            {
               logger_.DebugFormat("ignoring extraneous request to drop binding to exchange '{0}', routing-key '{1}'",
                  exchangeName, routingKey);
            }
            return true;
         }
         catch (System.Exception ex)
         {
            logger_.Exception(ex);
         }
         return false;
      }

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// Check the exchanges to see if we should bind to or unbind from anything
      /// </summary>
      private bool ThreadPlanet()
      {
         try
         {
            // Extract the requested actions
            List<KeyValuePair<string, Dictionary<string, bool>>> actions;
            lock (wantExchangeNames_)
            {
               actions = wantExchangeNames_.ToList();
               wantExchangeNames_.Clear();
               wantPending_.Set();
            }

            // Bind or unbind according to the request that was queued
            return actions.Any(kvpAction =>
               {
                  string exchangeName = kvpAction.Key;
                  channel_.ExchangeDeclare(exchangeName, ExchangeType.Topic, false, true, null);
                  kvpAction.Value.All(kvp => kvp.Value ?
                     ThreadBindRoutingKey(exchangeName, kvp.Key) : ThreadUnbindRoutingKey(exchangeName, kvp.Key)
                     );
                  return true;
               });
         }
         catch (System.Exception ex)
         {
            logger_.Exception(ex);
         }
         return false;
      }

      ////////////////////////////////////////////////////////////////////////
      /// <summary>
      /// 
      /// </summary>
      private bool ThreadResponse()
      {
         try
         {
            BasicDeliverEventArgs response = null;
            if (null != consumer_)
            {
               response = (BasicDeliverEventArgs)consumer_.Queue.DequeueNoWait(null);
            }
            if (null == response)
            {
               // Nothing to do
               return false;
            }

            // Ignore messages that we sent and that have been reflected back to us by RabbitMQ
            // And ignore messages that arrive after we have logged out
            if (senderId_ == response.BasicProperties.ReplyTo)
            {
               return false;
            }

            // Decode the message
            if (null != EventReceivedMessage)
            {
               using (MemoryStream stream = new MemoryStream(response.Body))
               {
                  // The body contains an encoding of an IMessage object, whose type name
                  // has been passed in the BasicProperties.ContentType (e.g. "Interfaces.IHeartbeat")
                  // The loader knows how to create such objects by name, so ...
                  string msgType = response.BasicProperties.ContentType;
                  IMessage msg = Program.Loader.Create(msgType) as IMessage;
                  if (null == msg)
                  {
                     logger_.WarnFormat("ignoring unknown/unexpected message type '{0}'",
                        msgType);
                  }
                  else if (msg.Deserialize(stream))
                  {
                     // Output the message to the debug log unless it's something boring
                     if (null == (msg as IHeartbeat))
                     {
                        logger_.DebugFormat("received message type '{0}' from {1}",
                           msg.DebugName, msg.SenderId);
                     }
                     ReceivedMessageEventArgs eventArgs = new ReceivedMessageEventArgs();
                     eventArgs.service = ExchangeNames.Broadcast;
                     eventArgs.mnemonic = response.BasicProperties.MessageId;
                     eventArgs.message = msg;
                     EventReceivedMessage(this, eventArgs);
                     return true;
                  }
               }
            }
            logger_.WarnFormat("ignoring message; there is no client listening");
         }
         catch (System.Exception ex)
         {
            logger_.Exception(ex);
            ThreadDisconnect();
         }
         return false;
      }

      //////////////////////////////////////////////////////////////////////////
      //
      private void ThreadHandleBasicReturn(IModel channel, BasicReturnEventArgs args)
      {
         try
         {
            logger_.WarnFormat("Message returned from exchange '{0}', key '{1}', code={2}, text='{3}'",
               args.Exchange, args.RoutingKey, args.ReplyCode, args.ReplyText);
            
         }
         catch (System.Exception ex)
         {
            logger_.Exception(ex);
            ThreadDisconnect();
         }
      }
   }
}
